package cn.mengfly.testnrz.reactor;

import cn.mengfly.testnrz.nio.NioDemoConfig;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;

public class TestReactor {

    static class EchoServerReactor implements Runnable {

        Selector selector;
        ServerSocketChannel serverSocket;

        public EchoServerReactor() throws IOException {
            // 获取Socket
            serverSocket = ServerSocketChannel.open();
            serverSocket.socket().bind(new InetSocketAddress(NioDemoConfig.SOCKET_PORT));
            serverSocket.configureBlocking(false);
            selector = Selector.open();

            // 绑定Selector
            SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
            sk.attach(new AcceptorHandler(serverSocket, selector));
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey sk : selectionKeys) {
                        dispatch(sk);
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private void dispatch(SelectionKey sk) {
            Runnable handler = (Runnable) sk.attachment();
            if (handler != null) {
                handler.run();
            }
        }
    }

    static class AcceptorHandler implements Runnable {
        private final ServerSocketChannel serverSocket;
        private final Selector selector;

        AcceptorHandler(ServerSocketChannel serverSocketChannel, Selector selector) {
            this.serverSocket = serverSocketChannel;
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                if (channel != null) {
                    System.out.println("新的连接接入了...");
                    new EchoHandler(channel, selector);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    static class EchoHandler implements Runnable {
        final SocketChannel channel;
        final SelectionKey sk;

        EchoHandler(SocketChannel c, Selector selector) throws IOException {
            channel = c;
            channel.configureBlocking(false);
            sk = channel.register(selector, SelectionKey.OP_READ);
            sk.attach(this);
            sk.interestOps(SelectionKey.OP_READ);
            selector.wakeup();
        }


        @Override
        public void run() {
            // 处理输入输出
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int length;
            try {
                while ((length = channel.read(buffer)) > 0) {
                    System.out.println(new String(buffer.array(), 0, length));
                    buffer.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
                try {
                    channel.close();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }

        }
    }

    public static void main(String[] args) throws IOException {
        new Thread(new EchoServerReactor()).start();
    }

}
